package mr.diff;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.functor.UnaryFunction;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import util.DiffComputation;
import util.Parsable;
import util.functors.AccociBinaryOP;
import util.functors.BinaryOperator;

public class DiffMapRed<T0 extends Parsable<T0>, T1 extends Parsable<T1>, T2 extends Parsable<T2>, MSG extends Writable>
		implements Mapper<LongWritable, Text, LongWritable, Text>,
		Reducer<LongWritable, Text, Text, NullWritable> {

	protected static final String ELE_Delim = " ";
	public T1 cons_c;
	protected DiffComputation<T0, T1, T2> diff;

	protected UnaryFunction<T1, T2> g;

	protected List<MSG> messageBox;
	protected int myId;
	protected String myName;
	protected AccociBinaryOP<T2> oplus;
	protected AccociBinaryOP<T1> otimes;
	protected BinaryOperator<T0, T1, T2> p;

	protected T1 otimsAcc;
	
	protected UnaryFunction<T0, T1> q;
	
	protected long chunkey;
	protected long nextChunkey;

	public DiffMapRed(final DiffComputation<T0, T1, T2> di) {
		this.diff = di;
		this.oplus = diff.oplus;
		this.otimes = diff.otimes;

		this.g = diff.g;
		this.p = diff.p;
		this.q = diff.q;
		this.cons_c = diff.init;
		
		messageBox = new ArrayList<MSG>();
		chunkey = -100;
	}

	
	public long getChunkey(){
		return chunkey;
	}
	
	/**
	 * get (and also set) the chunk key (and nextChunkey)
	 */
	public long getChunkey(Reporter reporter){
		String spl = reporter.getInputSplit().toString();
		if(spl != null && spl.length() > 0){
			String[] offset = spl.substring(spl.lastIndexOf(":")+1).split("\\+");
			chunkey = Long.parseLong(offset[0]);
			nextChunkey = Long.parseLong(offset[1]);
		}
		return chunkey;
	}

	public boolean isMapExectuted(){
		 if (chunkey >=-1) 
			 return true;
		 else
			 return false;
	}
	
	@Override
	public void close() throws IOException {
	}

	@Override
	public void configure(JobConf job) {
	}

	@Override
	public void map(LongWritable key, Text value,
			OutputCollector<LongWritable, Text> output, Reporter reporter)
			throws IOException {
	}

	@Override
	public void reduce(LongWritable key, Iterator<Text> values,
			OutputCollector<Text, NullWritable> output, Reporter reporter)
			throws IOException {

	}

}
